package demo.test.qsc.com.rxjavademo;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.TextView;
import android.widget.Toast;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;

public class MainActivity extends AppCompatActivity {
    final static String TAG = "MainActivity";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        TextView mBegin = findViewById(R.id.tv_begin);
        mBegin.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                Toast.makeText(MainActivity.this, "sssssss", Toast.LENGTH_SHORT).show();
//                observableSet();
                  observablechain(); //连式调用；
            }
        });
    }

    private void observablechain() {
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
//                Log.i(TAG,"emitter的值1");
//                e.onNext(1);
//                Log.i(TAG,"emitter的值2");
//                e.onNext(2);
//                Log.i(TAG,"emitter的值3");
//                e.onNext(3);
//                e.onComplete();
//                Log.i(TAG,"emitter的值4");
//                e.onNext(4);
//            }
//        }).subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Exception {
//                Log.i(TAG,"onNext"+integer);
//            }
//        });

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Log.i(TAG,"emitter的值1");
                e.onNext(1);
                Log.i(TAG,"emitter的值2");
                e.onNext(2);
                Log.i(TAG,"emitter的值3");
                e.onNext(3);
                e.onComplete();
                Log.i(TAG,"emitter的值4");
                e.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            private int i;
            private Disposable mDisposable;
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG,"subscribe");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                Log.i(TAG,"值是value");
                i++;
                if(i == 2){
                    Log.i(TAG,"dispose执行");
                    mDisposable.dispose(); // 结束任务传输切断水源但是 3，4 依然执行。
                    Log.i(TAG,"isDisposed"+mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG,"异常为"+e);
            }

            @Override
            public void onComplete() {
                Log.i(TAG,"结果已完成");
            }
        });
    }

//    private void observableSet() {
//        //创建 一个上游 Observable
//        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
//
//                e.onNext(1);
//                e.onNext(2);
//                e.onNext(3);
//                e.onNext(4);
//                e.onComplete();
//            }
//        });
//
//        //创建一个下游 Observer
//
//        Observer<Integer> observer = new Observer<Integer>() {
//            @Override
//            public void onSubscribe(Disposable d) {
//                Log.i(TAG,"subscribe方法");
//            }
//
//            @Override
//            public void onNext(Integer value) {
//                Log.i(TAG,"next传递过来的数值"+value);
//            }
//
//            @Override
//            public void onError(Throwable e) {
//                Log.i(TAG,"e错误提示"+ e);
//            }
//
//            @Override
//            public void onComplete() {
//                Log.i(TAG,"任务完成");
//
//            }
//        };
//        //建立链接
//        observable.subscribe(observer);
//    }
}
























